package com.sh.data.engine.domain.normalization.modeldesign.model.service.impl;

import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.domain.normalization.modeldesign.model.model.domain.CreateTableDomain;
import com.sh.data.engine.domain.normalization.modeldesign.model.model.domain.ModelDomain;
import com.sh.data.engine.domain.normalization.modeldesign.model.model.param.SaveModelParam;
import com.sh.data.engine.domain.normalization.modeldesign.model.service.TableOperationStrategy;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.shims.sql.builder.DDLBuilder;
import com.sh.data.engine.domain.shims.sql.ddl.AddTableFieldParam;
import com.sh.data.engine.domain.shims.sql.ddl.CreateTableParam;
import lombok.extern.slf4j.Slf4j;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class HiveTableOperation implements TableOperationStrategy {

    /**
     * 获取字段列表
     *
     * @param dbName
     * @param tableName
     * @param dbManager
     * @return
     */
    @Override
    public List<ModelDomain.Field> getTableField(
        String dbName, String schema, String tableName, BaseDbManager dbManager) {
        List<ModelDomain.Field> resultFields = new ArrayList<>();
        try {
            // connection已经在方法内close了
            List<FieldInfoDomain> fieldList = dbManager.getFieldList(dbName, tableName);
            for (FieldInfoDomain dbField : fieldList) {
                ModelDomain.Field resultField = new ModelDomain.Field();
                resultField.setFieldName(dbField.getFieldName());
                resultField.setFieldDesc(dbField.getComment());
                resultField.setFieldType(dbField.getFieldType());
                resultField.setIsPartition(dbField.getIsPartition());
                resultFields.add(resultField);
            }
        } catch (SQLException e) {
            log.error("Hive模式,hive 数据库:{} 查询:{} 字段异常:{}", dbName, tableName, e.getMessage());
            closeDbManager(dbManager);
            throw new BusinessException(
                "Hive模式下: 数据库:" + dbName + " 获取表字段:" + tableName + " 异常:" + e.getMessage());
        }
        return resultFields;
    }

    /**
     * hive table
     *
     * @param createTableDomain
     */
    @Override
    public boolean createTable(CreateTableDomain createTableDomain) {
        String tableName = createTableDomain.getTableName();
        BaseDbManager dbManager = createTableDomain.getDbManager();
        String dbName = createTableDomain.getDbName();
        String schema = createTableDomain.getSchemaName();
        List<SaveModelParam.Field> fieldList = createTableDomain.getFieldList();
        // hive表参数
        CreateTableParam createTableParam = new CreateTableParam();
        // 表名
        createTableParam.setTableName(tableName);
        createTableParam.setTableComment(createTableDomain.getTableComment());
        // 字段列表
        ArrayList<CreateTableParam.FieldInfo> generalFields = new ArrayList<>();
        ArrayList<CreateTableParam.FieldInfo> partitionFields = new ArrayList<>();
        for (SaveModelParam.Field requestField : fieldList) {
            CreateTableParam.FieldInfo fieldInfo = new CreateTableParam.FieldInfo();
            fieldInfo.setFieldName(requestField.getFieldName());
            // 设置为hive支持的字段类型
            fieldInfo.setFieldType(requestField.getFieldType());
            fieldInfo.setComment(requestField.getFieldDesc());
            // 分区字段单独放
            if (requestField.getIsPartition()) {
                partitionFields.add(fieldInfo);
            } else {
                generalFields.add(fieldInfo);
            }
        }
        if (generalFields.size() == 0 && partitionFields.size() > 0) {
            throw new BusinessException("必须包含至少一个非分区字段");
        }
        createTableParam.setFieldInfoList(generalFields);
        createTableParam.setPartitionFieldList(partitionFields);
        createTableParam.setExternal(false);

        try (DDLBuilder hiveDDLBuilder = dbManager.getDDLBuilder()) {
            hiveDDLBuilder.create(createTableParam).execute();
            return true;
        } catch (Exception e) {
            log.error("Hive模式下: 数据库:{} 创建hive表:{} 异常: {}", schema, tableName, e.getMessage());
            closeDbManager(dbManager);
            return false;
        }
    }

    @Override
    public boolean editFormTable(
        String dbName,
        String schema,
        ModelDomain dbModel,
        SaveModelParam saveModelParam,
        BaseDbManager dbManager) {
        String newTableName = saveModelParam.getEnName();
        TableInfoDomain tableInfoDomain =
            this.getTableInfoDomain(dbName, null, dbModel.getEnName(), dbManager);
        if (null == tableInfoDomain) {
            throw new BusinessException("检查数据库中表是否存在");
        }
        try (DDLBuilder ddlBuilder = dbManager.getDDLBuilder()) {
            // rename
            if (!Objects.equals(dbModel.getEnName(), newTableName)) {
                List<String> tableNameList = dbManager.getTableList(dbName, schema).stream()
                    .map(TableInfoDomain::getTableName).collect(Collectors.toList());
                if (tableNameList.contains(newTableName)) {
                    throw new BusinessException("存在相同模型表名");
                }
                ddlBuilder.reset().alterTableName(dbName, dbModel.getEnName(), newTableName).execute();
            }
            // comment
            if (!Objects.equals(dbModel.getCnName(), saveModelParam.getCnName())) {
                ddlBuilder
                    .reset()
                    .alterTableComment(dbName, newTableName, saveModelParam.getCnName())
                    .execute();
            }

            // TODO 富文本支持改变
            //    if (tableInfoDomain.isPartitionTable() != saveModelParam.isSupportRtf()) {
            //      // 重建表,并且修改storage格式
            //      if (null == createTableParam) {
            //        createTableParam = new CreateTableParam();
            //      }
            //      createTableParam.setStoreType(saveModelParam.isSupportRtf() ? "ORC" : "TEXTFILE");
            //    }
            List<SaveModelParam.Field> fields = saveModelParam.getFields();
            List<AddTableFieldParam> newTableFields = new ArrayList<>();
            for (SaveModelParam.Field field : fields) {
                if (field.getIsPartition()) {
                    continue;
                }
                AddTableFieldParam addTableFieldParam = new AddTableFieldParam();
                addTableFieldParam.setFieldName(field.getFieldName());
                addTableFieldParam.setFieldType(field.getFieldType());
                addTableFieldParam.setComment(field.getFieldDesc());
                newTableFields.add(addTableFieldParam);
            }
            ddlBuilder.reset().replaceTableFields(dbName, newTableName, newTableFields).execute();
            return true;
        } catch (Exception e) {
            log.error(
                "Hive模式下: 数据库:{} 编辑模型:{}, 表:{} 异常:{}",
                dbName,
                dbModel.getId(),
                dbModel.getEnName(),
                e.getMessage());
        }
        return false;
    }

    @Override
    public boolean isTableNameExist(
        String dbName, String schema, String tableName, BaseDbManager dbManager) {
        try {
            boolean hiveTableNameExist = dbManager.isTableNameExist(dbName, tableName);
            if (hiveTableNameExist) {
                log.error("Hive模式下检查 hive 数据源 {} 表{}已存在", dbName, tableName);
                return true;
            }
        } catch (SQLException e) {
            log.error("Hive模式下检查 hive 数据源 {} 表{}是否存在时,发生异常: {}", dbName, tableName, e.getMessage());
            closeDbManager(dbManager);
            throw new BusinessException(
                "Hive模式下检查 hive 数据源 " + dbName + " 表" + tableName + "是否存在时,发生异常: " + e.getMessage());
        }
        return false;
    }

    @Override
    public TableInfoDomain getTableInfoDomain(
        String dbName, String schema, String tableName, BaseDbManager dbManager) {
        try {
            return dbManager.getTableInfoByTableName(dbName, tableName);
        } catch (SQLException e) {
            log.error(e.getMessage());
        }
        return null;
    }

    /**
     * 解析字段变更
     *
     * @param dbModel
     * @param saveModelParam
     * @return
     */
    @Override
    public ModelServiceImpl.FieldsParseResult parseFields(
        ModelDomain dbModel, SaveModelParam saveModelParam) {
        List<ModelDomain.Field> dbFields = dbModel.getFields();
        Map<String, ModelDomain.Field> dbFieldsMap =
            dbFields.stream()
                .collect(Collectors.toMap(ModelDomain.Field::getFieldName, Function.identity()));
        List<SaveModelParam.Field> paramFields = saveModelParam.getFields();
        Map<String, SaveModelParam.Field> paramFieldsMap =
            paramFields.stream()
                .collect(Collectors.toMap(SaveModelParam.Field::getFieldName, Function.identity()));
        // 新增的字段
        List<SaveModelParam.Field> newFieldList = new ArrayList();
        // 更新的字段
        List<SaveModelParam.Field> updateFieldList = new ArrayList();
        // 删除的字段
        List<SaveModelParam.Field> deleteFieldList = new ArrayList();
        for (SaveModelParam.Field paramField : paramFields) {
            if (!dbFieldsMap.containsKey(paramField.getFieldName())) {
                newFieldList.add(paramField);
            }
        }
        for (ModelDomain.Field dbField : dbFields) {
            if (paramFieldsMap.containsKey(dbField.getFieldName())) {
                SaveModelParam.Field paramField = paramFieldsMap.get(dbField.getFieldName());
                boolean sameDescAndType =
                    Objects.equals(dbField.getFieldDesc(), paramField.getFieldDesc())
                        && (Objects.equals(dbField.getFieldType(), paramField.getFieldType()));
                boolean sameField = Objects.equals(dbField.getIsPartition(), paramField.getIsPartition());
                if (!sameDescAndType || !sameField) {
                    SaveModelParam.Field updateField = new SaveModelParam.Field();
                    updateField.setFieldName(paramField.getFieldName());
                    updateField.setFieldDesc(paramField.getFieldDesc());
                    updateField.setFieldType(paramField.getFieldType());
                    updateField.setRequired(paramField.getRequired());
                    updateField.setIsPartition(paramField.getIsPartition());
                    updateFieldList.add(updateField);
                }
            } else {
                SaveModelParam.Field deleteItem = new SaveModelParam.Field();
                deleteItem.setFieldName(dbField.getFieldName());
                deleteFieldList.add(deleteItem);
            }
        }
        return ModelServiceImpl.FieldsParseResult.builder()
            .newFieldList(newFieldList)
            .updateFieldList(updateFieldList)
            .deleteFieldList(deleteFieldList)
            .build();
    }
}
